package com.dmp.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.dmp.logger.ILogger;

import java.util.Map;

/**
 * @ClassName BaseStringBolt
 * @Description 数据流为字符串流
 *
 * @author sunzy
 * @date 2016年2月24日 下午2:58:51 
 * @version 1.0 
 * 
 */
public abstract class BaseStringBolt extends BaseRichBolt {
	private static final long serialVersionUID = 1L;
	protected OutputCollector collector;
	protected ILogger logger;

	@Override
	public void execute(Tuple input) {
		try {
			String logStr = (String) input.getString(0);
			logStr = logStr.trim();

			String str = this.process(logStr);
			if (str != null) {
				collector.emit(new Values(str));
			}

			collector.ack(input);
		} catch (Exception e) {
			// e.printStackTrace();
			collector.fail(input);
		}
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map mapL, TopologyContext tpContext,
			OutputCollector collector) {
		this.collector = collector;
		prepare();
	}

	public void prepare() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("resultMap", "keyword"));
	}

	public String process(String str) {
		System.out.println("bolt:" + str);
		return null;
	}

}
